/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.scheduler.specific.sge.sched;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket;

import org.apache.log4j.Logger;

import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;

class SGEOperatord {

	private static final String OPED_PATH = "sge_operatord";
	private static final String OPED_HOSTNAME = "localhost";
	private static final int OPED_PORT = 16000;
	private static final int MAX_TRY = 10;
	private static final int START_WAIT = 300;	// [msec]
	private static final int CONNECT_WAIT = 5000;	// [msec]

	private Socket socket = null;
	private static final int NOT_CONNECTED = 0;
	private static int opedPort = NOT_CONNECTED;

	protected static final Logger logger = Logger.getLogger(SGEOperatord.class);
	private static final String STREAM_CHARSET = "US-ASCII";

	private static int getOpedPort() {
		return opedPort;
	}

	private static void setOpedPort(int port) {
		opedPort = port;
	}

	public SGEOperatord() throws SchedulerException {

		if (getOpedPort() == NOT_CONNECTED) {
			restart();
		} else {
			throw new SchedulerException(
					"sge_operatord is already started at port "
					+ getOpedPort() + ", cannot restart sge_operatord");
		}

	}

	private void restart() throws SchedulerException {

		checkStopped();
		int port = start();
		setOpedPort(port);

	}

	private void checkStopped() {

		logger.debug("start to check whether " + OPED_PATH + " is runnning");
		for (int port = OPED_PORT; port < OPED_PORT + MAX_TRY; port++) {
			try {
				connect(port);
				// oped is running now, shutdown it.
				shutdown();
			} catch (SchedulerException e) {
				// oped is not running now, it's OK.
			}
		}
		logger.debug("check finished");

	}

	private int start() throws SchedulerException {

		Runtime runtime = Runtime.getRuntime();
		for (int port = OPED_PORT; port < OPED_PORT + MAX_TRY; port++) {
			String[] cmd = { OPED_PATH, Integer.toString(port) };
			Process proc;
			try {
				proc = runtime.exec(cmd);
			} catch (Exception e) {
				// ignore, try next port
				continue;
			}

			try {
				Thread.sleep(START_WAIT);
			} catch (InterruptedException e1) {
				// ignore
			}

			try {
				proc.exitValue();
				// proccess is dead now, maybe failed to port open
			} catch (IllegalThreadStateException e) {
				// process is running now, it's OK.
				logger.debug("sge_operatord at port " + port + " started");
				return port;
			}
		}

		throw new SchedulerException("Cannot start " + OPED_PATH);

	}

	private Socket connect(int port) throws SchedulerException {

		try {
			logger.debug("try to connect sge_operatord at port " + port);
			socket = new Socket();
			socket.connect(new InetSocketAddress(OPED_HOSTNAME, port), CONNECT_WAIT);
			logger.debug("connected sge_operatord at port " + port);
			return socket;
		} catch (IOException e) {
			logger.debug("Cannot connect to sge_operatord at port " + port);
			socket = null;
			throw new SchedulerException("Cannot connect to sge_operatord at port " + port);
		}

	}

	public synchronized Socket connect() throws SchedulerException {

		if (isConnected()) {
			/*
			 * NOTE: sge_operatord doesn't support multi session.
			 */
			throw new SchedulerException("Another session is alive, cannot connect");
		}

		try {
			return connect(getOpedPort());
		} catch (SchedulerException e) {
			restart();
			return connect(getOpedPort());
		}

	}

	public synchronized boolean isConnected() {

		return (socket != null) ? socket.isConnected() : false;

	}

	private void close() {
		assert(socket != null);
		try {
			socket.close();
		} catch (IOException e) {
			logger.info("cannot close socket at port " + getOpedPort(), e);
		}
		socket = null;
	}

	public synchronized void disconnect() {

		if (isConnected()) {
			try {
				sendRequest("Bye");
			} catch (SchedulerException e) {
				logger.info("cannot disconnect at port " + getOpedPort(), e);
			} finally {
				close();
				logger.debug("disconnected sge_operatord at port " + getOpedPort());
			}
			// NOTE: sge_operatord is still running...
		}

	}

	public synchronized void shutdown() {

		if (isConnected()) {
			int port = socket.getPort();
			try {
				sendRequest("Shutdown");
			} catch (SchedulerException e) {
				logger.info("cannot shutdown at port " + getOpedPort(), e);
			} finally {
				close();
				logger.debug("sge_operatord at port " + port + " shutdowned");
			}
			// NOTE: sge_operatord was shutdowned, not running now.
			setOpedPort(NOT_CONNECTED);
		}

	}

	public synchronized void sendRequest(String data) throws SchedulerException {

		try {
			BufferedWriter writer = new BufferedWriter(
					new OutputStreamWriter(socket.getOutputStream(), STREAM_CHARSET));
			writer.write(data);
			writer.newLine();
			writer.flush();
			logger.debug("Sended: " + data);
		} catch (IOException e) {
			logger.debug(e);
			throw new SchedulerException(e.getMessage());
		}

	}

	public synchronized String receiveReply() throws SchedulerException {

		String data;
		try {
			BufferedReader reader = new BufferedReader(
				new InputStreamReader(socket.getInputStream(), STREAM_CHARSET));
			// NOTE: readLine() drops linefeed, data.trim() isn't necessary
			data = reader.readLine();
		} catch (IOException e) {
			logger.debug(e);
			throw new SchedulerException(e.getMessage());
		}

		logger.debug("Received: " + data);
		return data;

	}

}
